/*
 * RtcpRtpSocket.cpp
 *
 *  Created on: 2018年5月14日
 *      Author: zhengboyuan
 */

#include "RtcpRtpSocket.h"
#include "JRTPSession.h"
#include "RtpPort.h"
#include "RtpConst.h"
#include "RtpHeader.h"
#include "CLog.h"
#include "TFileUtil.h"

namespace av
{

RtcpRtpSocket::RtcpRtpSocket():
		m_working()
{

}

RtcpRtpSocket::~RtcpRtpSocket()
{
	close();
}

bool RtcpRtpSocket::open(const char* ip, int port)
{
	if (port == 0)
	{
		port = RtpPort::make();
	}

	for (size_t i = 0; i < RtpPort::count(); ++ i)
	{
		if (openRtpSession(port))
		{
			break;
		}
		else
		{
			port = RtpPort::make();
		}
	}

	if (!m_session)
	{
		return false;
	}

	m_localAddr.set(ip, port);

	m_working = true;

	CLog::debug("RtcpRtpSocket::open. port: %d\n", port);

	return 0;
}

void RtcpRtpSocket::close()
{
	m_working = false;

	closeRtpSession();
}

bool RtcpRtpSocket::isOpen()
{
	return m_session.get() != NULL;
}

NetAddress RtcpRtpSocket::getLocalAddr()
{
	return m_localAddr;
}

bool RtcpRtpSocket::send(RtpPacket& packet, int ms)
{
	if (!m_session)
	{
		return false;
	}

	poll();

	int rc = m_session->SendPacketEx(packet.data, packet.size, packet.pt, packet.mark, packet.ts);

    int delayMs = m_sendDelay.send(packet.size);
    m_sendDelay.delay(delayMs);

	return (rc == 0);
}

bool RtcpRtpSocket::recv(RtpPacketHandler handler, int ms)
{
	if (!m_session)
	{
		return false;
	}

	poll();

	readAndHandle(ms);

	while (m_rtpCache.isReady() && m_working)
	{
		RTPPacketPtr rtpPacket = m_rtpCache.pop();
		if (rtpPacket)
		{
			RtpPacket pkt;
			memset(&pkt, 0, sizeof(pkt));
			pkt.pt = rtpPacket->GetPayloadType();
			pkt.mark = rtpPacket->HasMarker();
			pkt.ts = rtpPacket->GetTimestamp();
			pkt.data = rtpPacket->GetPayloadData();
			pkt.size = rtpPacket->GetPayloadLength();

//			CLog::debug("incoming rtp packet. size: %4d, seq: %6d, eseq: %d\n",
//				rtpPacket->GetPayloadLength(), rtpPacket->GetSequenceNumber(), rtpPacket->GetExtendedSequenceNumber());

			handler(pkt);

			//comn::FileUtil::write(pkt.data, pkt.size, "out.ps", true);

			m_session->DeletePacket(rtpPacket);
		}
		else
		{
			break;
		}
	}

	return true;
}

bool RtcpRtpSocket::isConnected()
{
	return isOpen();
}

void RtcpRtpSocket::setSsrc(uint32_t ssrc)
{
	BasicRtpSocket::setSsrc(ssrc);

	if (m_session)
	{
		m_session->SetSSRC(ssrc);
	}
}

bool RtcpRtpSocket::setPeer(const char* ip, int port)
{
	BasicRtpSocket::setPeer(ip, port);

	CLog::info("RtcpRtpSocket::setPeer %s:%d. JRTPSession: %p\n", ip, port, m_session.get());

	addDestination(ip, port);
	return true;
}

bool RtcpRtpSocket::openRtpSession(uint16_t port)
{
    m_session.reset(new JRTPSession(NULL));

    RTPUDPv4TransmissionParams transparams;
    RTPSessionParams sessparams;
    sessparams.SetOwnTimestampUnit(1.0/90000.0);
    sessparams.SetAcceptOwnPackets(true);
    sessparams.SetUsePollThread(false);

	if (m_ssrc != 0)
	{
		sessparams.SetPredefinedSSRC(m_ssrc);
		sessparams.SetUsePredefinedSSRC(true);
	}

    transparams.SetRTPReceiveBuffer(RtpSocket::SEND_BUFFER_SIZE);
    transparams.SetRTPSendBuffer(RtpSocket::RECV_BUFFER_SIZE);

    transparams.SetPortbase(port);
    int ret = m_session->Create(sessparams, &transparams);
    if (ret != 0)
    {
        m_session.reset();
    }
    else
    {
        //m_session->SetDefaultPayloadType(m_payload);
        m_session->SetMaximumPacketSize(RtpConst::RTP_MAX_SIZE + sizeof(RTP_FIXED_HEADER)*2);
        m_session->SetDefaultTimestampIncrement(0);
        m_session->SetDefaultMark(false);
    }
    return (ret == 0);
}

void RtcpRtpSocket::closeRtpSession()
{
	comn::AutoCritSec lock(m_cs);

    if (m_session)
    {
    	abort();

    	RTPTime t(1, 0);
        std::string reason("session closed");
        m_session->BYEDestroy(t, (void*)reason.c_str(), reason.size());

        clearCache();

        m_session.reset();
    }
}

void RtcpRtpSocket::poll()
{
	comn::AutoCritSec lock(m_cs);
	m_session->Poll();
}

void RtcpRtpSocket::readAndHandle(int ms)
{
    bool dataAvailable = true;
    RTPTime timeout(ms/1000, ms % 1000);
    m_session->WaitForIncomingData(timeout, &dataAvailable);

    m_session->BeginDataAccess();

    bool working = m_session->GotoFirstSourceWithData();
	//CLog::debug("%s GotoFirstSourceWithData. %d\n", CLog::getShortTime(), working);

    while (working && m_working)
    {
        //process packet
        RTPPacket* pPacket = NULL;
        pPacket = m_session->GetNextPacket();
        while (pPacket)
        {
            onRtpPacket(pPacket);

            if (!m_working)
            {
                break;
            }

            pPacket = m_session->GetNextPacket();
        }

        working = m_session->GotoNextSourceWithData();
    }

    m_session->EndDataAccess();
}

void RtcpRtpSocket::onRtpPacket(jrtplib::RTPPacket* pPacket)
{
    m_rtpCache.push(pPacket);
}

void RtcpRtpSocket::clearCache()
{
    m_rtpCache.clear(m_session.get());
}

void RtcpRtpSocket::addDestination(const std::string& ip, int port)
{
	if (!m_session)
	{
		return;
	}

	uint32_t ipAddr = inet_addr(ip.c_str());
	ipAddr = ntohl(ipAddr);
	RTPIPv4Address dest(ipAddr, port);
	m_session->AddDestination(dest);
}

void RtcpRtpSocket::abort()
{
    if (m_session)
    {
        m_session->AbortWait();
    }
}


} /* namespace av */
